// -*-c++-*-

#include <sys/stat.h>
#include <fcntl.h>


//#include <tr1/tuple>
//#include <tr1/functional>
//#include <tr1/memory>
#include <sstream>
#include <algorithm>

#include <boost/foreach.hpp>
#include <boost/bind.hpp>

#include <xthread/guard.hpp>

#include <xbase_utility.hpp>
#include <dynamic_buffer.hpp>
#include <fs.hpp>
#include <config.hpp>

#include "tablet_server_configuration.hpp"
#include "log_thread.hpp"
#include "naming_strategy.hpp"
#include "group_scanner.hpp"
#include "global.hpp"
#include "table_schema.hpp"
#include "table_schema_manager.hpp"
#include "kv_scanner.hpp"

#include "tablet.hpp"

static inline
char calc_avg(char a, char b)
{
	const unsigned int ia = static_cast<unsigned int>(static_cast<unsigned char>(a));
	const unsigned int ib = static_cast<unsigned int>(static_cast<unsigned char>(b));
	const unsigned int ir = (ia + ib) >> 1;
	return static_cast<char>(static_cast<unsigned char>(ir));
}

static inline
std::string calc_split_row(const std::string &start_row,
			   const std::string &end_row)
{
	if(start_row.empty() || end_row.empty())
	{
		const char ch1 = (start_row.empty() ? char(0) : start_row[0]);
		const char ch2 = (end_row.empty() ? char(255) : end_row[0]);
		const char ch = calc_avg(ch1, ch2);
		return std::string(1, ch);
	}
	else
	{
		// find first position not equal
		const std::size_t min_size = (std::min)(start_row.size(), end_row.size());
		std::size_t i = 0;
		for(; i < min_size; ++i)
		{
			if(start_row[i] != end_row[i])
				break;
		}
		const char ch1 = (start_row.size() > i ? start_row[i] : 0);
		const char ch2 = (end_row.size() > i ? end_row[i] : 0);
		const char ch = calc_avg(ch1, ch2);
		std::string mid_row(start_row, 0, i); // 共同前缀
		mid_row.append(1, ch);
		return mid_row;
	}
}

namespace xbase {

	tablet::pointer tablet::load(const std::string &p_root_path,
				     const tablet_info &p_info)
	{
		// first, check tablet path exists
		const fs::path _tablet_path = naming_strategy::tablet_path(p_root_path,
									   p_info.table_name,
									   p_info.tablet_id,
									   p_info.start_row,
									   p_info.end_row);
		std::vector<fs::file_info> _files;
		if (! fs::exists(_tablet_path.string()))
		{
			if(! fs::mkdir(_tablet_path.string()))
			{
				LOG(ERROR, "mkdir " << _tablet_path << " failed.");
				return tablet::pointer();
			}
		} else {
			if(! fs::list_files(_files, _tablet_path.string()))
			{
				LOG(ERROR, "list files in " << _tablet_path << " failed.");
				return tablet::pointer();
			}
		}

		table_schema::pointer _schema = global::the_schema_manager->get_schema(p_info.table_name);
		if (! _schema)
		{
			LOG(ERROR, "tablet::load table schema of '"
			    << p_info.table_name << "' is null.");
			return tablet::pointer();
		}
		
		pointer _tablet(new tablet(p_info.tablet_id,
					   p_info.table_name,
					   p_info.start_row,
					   p_info.end_row,
					   p_info.status));

		fs::path _oldlog_path = naming_strategy::oldlog_path(_tablet_path);
		bool _has_oldlog = false;
		
		// 按目录加载组，同时处理不是组的目录
		BOOST_FOREACH(fs::file_info _f, _files)
		{
			const std::string _group_name = naming_strategy::groupname_for_dirname(fs::get_name(_f));
			const table_schema::group_info::pointer group_info = _schema->get_group_info(_group_name);
			group::family_map_type families;
			_schema->list_family(families, _group_name);
			if(group_info) // 组存在
			{
				group::pointer __g = group::load(_tablet_path.string(),
								 _schema->table_name(),
								 _group_name,
								 _tablet->m_tablet_id,
								 _tablet->tablet_name(),
								 _tablet->start_row(),
								 _tablet->end_row(),
								 *group_info,
								 families);
				if (!__g) {
					LOG(WARN, "load group " << _group_name
					    << " from " << _tablet_path << " failed.");
					// as a failure
					return tablet::pointer();
				} else {
					_tablet->append(__g);
					LOG(INFO, "group " << _group_name << " loaded.");
				}
			}
			else if (fs::get_name(_f) == _oldlog_path.leaf())
			{
				_has_oldlog = true;
				LOG(TRACE, "group::load found oldlog file.");
			}
			else
			{
				// 如果组不存在，则按目录改为特殊名字。
				// 比如以".trash_"开始的名称，以后碰到
				// 这类的组名就直接跳过
				if (naming_strategy::is_trash_file(fs::get_name(_f))) {
					LOG(TRACE, "group::load skip trash file "
					    << fs::get_name(_f) << ".");
				} else {
					LOG(TRACE, "group::load " << fs::get_name(_f)
					    << "(" << _group_name << ") is not a valid group dir, trash it.");
					fs::rename((_tablet_path / fs::get_name(_f)).string(),
						   (_tablet_path / naming_strategy::trash_filename(fs::get_name(_f))).string());
				}
			}
		}

		// 检查是否所有组全部加载
		// 
		// 因为每个加载的组在schema中都有，所在数量相同就表示全部加载了
		bool _all_loaded = (_schema->group_number() == _tablet->group_number());
		if (!_all_loaded)
		{
			std::vector<std::string> __list;
			_schema->list_group(__list);
			BOOST_FOREACH(const std::string &__name, __list)
			{
				if (! _tablet->loaded(__name))
				{
					// 如果没有加载，说明没有这个组目录，因此创建它
					std::string _dirname = naming_strategy::group_dirname(__name);
					fs::path _path = _tablet_path / _dirname;
					if(! fs::mkdir(_path.string()))
					{
						LOG(ERROR, "create group dir " << _path << " failed.");
						return tablet::pointer();
					}
					else
					{
						group::family_map_type families;
						_schema->list_family(families, __name);
						group::pointer __g = group::load(_tablet_path.string(),
										 _schema->table_name(),
										 __name,
										 _tablet->m_tablet_id,
										 _tablet->tablet_name(),
										 _tablet->start_row(),
										 _tablet->end_row(),
										 *(_schema->get_group_info(__name)),
										 families);
						if (! __g) {
							LOG(ERROR, "tablet::load load new group "
							    << __name << " failed.");
							// as a failure
							return tablet::pointer();
						}
						_tablet->append(__g);
						LOG(TRACE, "created new dir for group " << __name);
					}
				}
			}
		}

		// load oldlog
		if (_has_oldlog)
		{
			LOG(TRACE, "tablet::load loading oldlog data from " << _oldlog_path);
			bool _ok = _tablet->load_oldlog(_oldlog_path.string());
			if(! _ok)
			{
				std::string dest_name = _oldlog_path.string();
				const std::string root_path = global::xbase_root_path().string();
				if((dest_name.size() > root_path.size()) &&
				   (dest_name.compare(0, root_path.size(), root_path) == 0))
				{
					// 将前缀去掉
					if(dest_name[root_path.size()] == '/')
						dest_name.erase(0, root_path.size() + 1); // 这是正常情况
					else
						dest_name.erase(0, root_path.size());
				}
				std::replace(dest_name.begin(), dest_name.end(), '/', '_');
				dest_name += '.';
				dest_name += boost::lexical_cast<std::string>(cur_micro_seconds());
				const fs::path dest_path =
					global::xbase_root_path() / ".trash" / dest_name;
				LOG_ENV(const bool ok_rename = ) fs::rename(_oldlog_path, dest_path);
				LOG(ERROR, "tablet::load failed: load oldlog " << _oldlog_path
				    << " failed, tablet = " << _tablet->tablet_name()
				    << ", move oldlog to .trash: " << dest_path);
				LOG_IF((!ok_rename), ERROR,
				       "but move failed, fs::errno = " << fs::get_errno());
				return tablet::pointer();
			}

			_ok = _tablet->recover_compact();
			if(! _ok)
			{
				LOG(ERROR, "tablet::load recover compact failed");
				return tablet::pointer();
			}

			// 待紧缩完成后再删除旧日志，否则若在紧缩过程中挂掉，数据就丢失了！
			fs::remove(_oldlog_path.string());
			LOG(TRACE, "tablet::load oldlog data loaded.");
		}

		// 注意，根子表不进行分裂！
		// check major compact if not root meta tablet
		// 这里不需要事先设置状态，因为没人会去使用这个子表
		_tablet->set_online(true); // 分裂需要子表处于上线状态
		bool _ok = _tablet->major_compact_or_split();
		if (! _ok) {
			LOG(ERROR, "tablet::load major_compact_or_split failed.");
			return tablet::pointer();
		}
		// set status
		if (_tablet->is_split())
		{
			_tablet->set_online(false);
		}
		else
		{
			_tablet->set_online(true);
		}

		return _tablet;
	}

	tablet::tablet(uint64_t p_tablet_id,
		       const std::string p_table_name,
		       const std::string p_start_row,
		       const std::string p_end_row,
		       uint16_t p_status)
		: m_compact_status(cs_null)
		, m_tablet_id(p_tablet_id)
		, m_table_name(p_table_name)
		, m_start_row(p_start_row)
		, m_end_row(p_end_row)
		, m_status(p_status)
		, m_is_compacting(false)
		, m_tablet_name(naming_strategy::tablet_name(p_tablet_id,
							     p_table_name,
							     p_start_row,
							     p_end_row))
	{
	}

	int tablet::add(const protocol::row_mutation_data_for_tabletserver &data,
			const boost::asio::const_buffer &raw_data,
			error::error_code &error_code)
	{
		//Guard<ReadWriteLock::ReadLock> guard(m_lock.read_lock());
		boost::shared_lock<boost::shared_mutex> s_lock(m_lock);
		if(! this->is_online()) // not online
		{
			LOG(DEBUG, "tablet::add tablet "
			    << this->tablet_name()
			    << " is not online");
			error_code = error::tablet_closed;
			return -1;
		}

		LOG_ENV(std::size_t record_count = 0);
		LOG_ENV(std::size_t data_bytes = 0);

		// 通过了写前的所有检查，创建日志条目，准备写日志
		std::vector<std::string> group_list;

		// make row block
		typedef std::pair<group::pointer, row_block::pointer> group_row_block;
		std::vector<group_row_block> row_blocks;
		row_blocks.reserve(data.m_groups.size());
		BOOST_FOREACH(const protocol::row_mutation_data_for_tabletserver::group &g, data.m_groups)
		{
			if(boost::asio::buffer_size(g.second) == 0) // skip empty data
				continue;

			// 这里去掉的检查依次有：
			// 行数据必须大于4字节，即有行名
			// 行名不能为空 - 否则会导致很多意料不到的错误。因为空行通常用来表示没有行。
			// 以及
			// 是否同一行记录
			// 列是否包含正则表达式
			// 列族是否存在
			//
			// 以及
			// 为新的数据设置时间戳（现在改到客户端来做，即所有数据都有时间戳了）

			// 如果组名存在，则必须获取到或创建组，因此组返回空是较严重的错误
			group::pointer p = get_or_add_group(g.first);
			if(! p)
			{
				LOG(ERROR, "tablet::add get or add group "
				    << g.first << " failed");
				error_code = error::group_not_exists;
				return -1;
			}
			if(p->too_huge())
			{
				LOG(DEBUG, "memory table too huge, group = "
				    << p->group_name() << ", tablet = "
				    << this->tablet_name());
				error_code = error::memtable_too_huge;
				notify_minor_compact();
				return -1;
			}
			
			const row_block::pointer row = make_row_block(g.second);
			LOG_ENV(record_count += row->block_index().size());
			LOG_ENV(data_bytes += row->data_bytes());

			row_blocks.push_back(group_row_block(p, row));
			group_list.push_back(g.first);
		}

		if(row_blocks.empty())
		{
			error_code = error::ok;
			return 0;
		}

		// 只检查一行即可
		const key_wrapper::string_type * const row_w = row_blocks.front().second->row_name();
		const std::string row = (std::string)*row_w;
		if(row.empty())
		{
			LOG(ERROR, "cannot write records into empty row, tablet_name = "
			    << this->tablet_name())
			error_code = error::bad_row_name;
			return -1;
		}
		if(! IS_IN_RANGE(row, m_start_row, m_end_row))
		{
			LOG(ERROR, "tablet::add row \""
			    << dump_string_char(row)
			    << "\" not in tablet range ["
			    << dump_string_char(m_start_row)
			    << ", "
			    << dump_string_char(m_end_row)
			    << ")");
			error_code = error::row_not_in_range;
			return -1;
		}

		//
		// 3. write log，先写日志
		//
		// 将写日志放到这里，按组分类并进行完所有检查之后！因为
		// 这些检查通过之后，写入内存表的操作几乎是不会失败的。
		// 所以这是写日志的最佳时机！若写日志成功，则可说明写操
		// 作几乎已经成功了。
		//
		// TODO: 
		// 否则，若万一后面的写操作失败了，向客户端返回失败，同
		// 时将该子表关闭，进行紧缩并卸载，这样日志中的数据就不
		// 会被恢复了。（现在不考虑这种情况，待实际测试无法保证
		// 后面写操作不成功的再说）
		// 

		const bool ret = global::the_log_manager->logger().log(
			this->tablet_name(),
			row,
			raw_data,
			group_list);
		if(! ret)
		{
			error_code = error::log_failed;
			LOG(ERROR, "tablet::add write log failed");
			return -1;
		}
		// 如果写入失败了，但此时又写内存表成功了，已经无法恢复
		// 了，只有在内存表紧缩之前退出程序。
		//
		// 但是，这样也无法保证写入的数据没有被写入到
		// SSTable中，如果先写日志的话，就会好一些。
		//assert(ret == 0);
		// int ret = write_log();
		// if(ret < 0) failed

		// 
		// 现在假定以下的操作不会失败；若测试发现有失败的情况的
		// 话，就需要将该子表关闭、紧缩、下线，总之，在保证数据
		// 不丢失的情况下，避免写入的日志被恢复，并要向客户端返
		// 回失败
		// 

		// 将所有缓冲区有数据的组的数据提交，作为严重错误，如果
		// 写的过程中失败不能失败，否则子表服务器只有退出，才能
		// 保证数据的完整性！
		//
		// 现在只有通过这种方式保证一次写入的数据要么都写成功，
		// 要么都不成功；但要保证写日志在成功写入内存表之后！

		bool need_minor_compact = false;
		BOOST_FOREACH(group_row_block &row, row_blocks)
		{
			row.first->add(row.second);
			need_minor_compact = (need_minor_compact 
					      || row.first->is_need_minor_compact_fastly());
		}
		// notify minor compact
		if(need_minor_compact)
		{
			notify_minor_compact();
		}

		error_code = error::ok;
		LOG(TRACE, "row \"" << dump_string_char(row)
		    << "\" written with "
		    << record_count << " records and "
		    << data_bytes << " bytes data, affected "
		    << row_blocks.size()
		    << " groups, raw group data size = "
		    << boost::asio::buffer_size(raw_data));
		return 0;
	}

	group::pointer tablet::get_or_add_group(const std::string &group)
	{
		group::pointer ptr; 
		{
			//Guard<ReadWriteLock::ReadLock> _guard(m_groups_lock.read_lock());
			boost::shared_lock<boost::shared_mutex> _guard(m_groups_lock);
			group_map::iterator _iter = m_groups.find(group);
			if (_iter != m_groups.end())
			{
				ptr = _iter->second;
			}
		}

		// TO_CONFIRM: 组不存在，但在表结构中有，可能是后来更新了表结构。所以这里最好创建该新组！
		// 这就是更改表结构中添加组的实现方式
		if(! ptr)
		{
			LOG(WARN, "tablet::add: group "
			    << group << " not found in tablet "
			    << this->tablet_name() << ", create it.");
			
			table_schema::pointer schema = this->get_schema();
			if(! schema)
			{
				LOG(ERROR, "tablet::get_or_add_group table schema of '"
				    << this->table_name() << "' is null.");
				return group::pointer();
			}
			table_schema::group_info::pointer group_info = schema->get_group_info(group);
			if(! group_info)
			{
				LOG(ERROR, "tablet::get_or_add_group group " << group
				    << " not exists in '"
				    << this->table_name() << "'");
				return group::pointer();
			}
			group::family_map_type families;
			schema->list_family(families, group);

			const fs::path _tablet_path = naming_strategy::tablet_path(global::xbase_root_path().string(),
										   m_table_name,
										   m_tablet_id,
										   m_start_row,
										   m_end_row);
			const fs::path _group_path = _tablet_path / naming_strategy::group_dirname(group);
			if(! fs::exists(_group_path))
			{
				if(! fs::mkdir(_group_path))
				{
					LOG(ERROR, "tablet::add create new group dir " << _group_path << " failed.");
					return group::pointer();
				}
			}
			
			group::pointer _g = group::load(_tablet_path.string(),
							this->table_name(),
							group,
							this->m_tablet_id,
							this->tablet_name(),
							this->start_row(),
							this->end_row(),
							*group_info,
							families);
			if (! _g)
			{
				LOG(ERROR, "tablet::add create new group " << group << " failed.");
				return group::pointer();
			}
			this->append(_g);
			LOG(DEBUG, "tablet::add new group " << group << " created and loaded.");
			ptr = _g;
		}
		return ptr;
	}

	tablet_scanner::pointer tablet::get_scanner(kv_scan_context::pointer p_context)
	{
		//Guard<ReadWriteLock::ReadLock> guard(m_lock.read_lock());
		boost::shared_lock<boost::shared_mutex> guard(m_lock);
		if(! this->is_online()) // not online
		{
			LOG(DEBUG, "tablet::add tablet "
			    << this->tablet_name()
			    << " is not online");
			return tablet_scanner::pointer();
		}

		tablet_scanner::pointer _scanner(new tablet_scanner(p_context,
								    this->shared_from_this()));

		// 这种方式的并行性比上面的方式要好吧

		// 1. 首先找出要扫描的组
		std::vector<group::pointer> _list;
		_list.reserve(p_context->groups().size());
		if(! p_context->is_contain_all_columns())
		{
			//Guard<ReadWriteLock::ReadLock> _guard(m_groups_lock.read_lock());
			boost::shared_lock<boost::shared_mutex> _guard(m_groups_lock);
			const std::vector<std::string>::const_iterator end = p_context->groups().end();
			std::vector<std::string>::const_iterator i = p_context->groups().begin();
			for(; i != end; ++ i)
			{
				const group_map::iterator iter = m_groups.find(*i);
				if(iter != m_groups.end())
				{
					_list.push_back(iter->second);
				}
				else
				{
					LOG(DEBUG, "tablet::get_scanner not group object "
					    << (*i) << " exists, skip it");
					// 这里应该是组在表结构中存在，
					// 但没有创建对象（即还没有给该
					// 组中写数据），因此不在这里创
					// 建该组，而是交给写操作来添
					// 加，这里只需忽略该组即可。
				}
			}
		}
		else		// take all group
		{
			//Guard<ReadWriteLock::ReadLock> _guard(m_groups_lock.read_lock());
			boost::shared_lock<boost::shared_mutex> _guard(m_groups_lock);
			BOOST_FOREACH(const group_map::value_type &_elem, m_groups)
			{
				_list.push_back(_elem.second);
			}
		}
		// 2. 只从要扫描的组中获取扫描器
		BOOST_FOREACH(group::pointer &_group, _list)
		{
			group_scanner::pointer __p = _group->get_scanner(p_context);
			if (!__p)
			{
				LOG(ERROR, "tablet::get_scanner failed in group "
				    << _group->group_name() << ".");
				return tablet_scanner::pointer();
			}
			if(__p->eos())
			{
				LOG(DEBUG, "tablet::get_scanner empty scanner in group "
				    << _group->group_name() << ", skip it");
			}
			else
			{
				_scanner->add(__p);
			}
		}
		return _scanner;
	}

	void tablet::append(const group::pointer &p_group)
	{
		//Guard<ReadWriteLock::WriteLock> _guard(m_groups_lock.write_lock());
		boost::unique_lock<boost::shared_mutex> _guard(m_groups_lock);
		std::pair<group_map::iterator, bool>
			ret = m_groups.insert(group_map::value_type(p_group->group_name(),
								    p_group));
		// DONE: 如果新加入的组原来就存在，则将其忽略即可
		if(ret.second == false)
		{
			LOG(ERROR, "tablet::append group "
			    << p_group->group_name()
			    << " already exists, skip the new one");
		}
	}
	
	table_schema::pointer tablet::get_schema() const
	{
		// double-checked lock
		if(m_schema.expired())
		{
			//Guard<FastLock> _guard(m_schema_lock);
			boost::mutex::scoped_lock _guard(m_schema_lock);
			if (m_schema.expired())
			{
				m_schema = global::the_schema_manager->get_schema(this->table_name());
			}
		}
		return m_schema.lock();
	}

	bool tablet::load_oldlog(const std::string &p_path)
	{
		fs::file_t _file = fs::open(p_path);
		if (_file == fs::BAD_FILE)
		{
			LOG(ERROR, "tablet::load_oldlog open file " << p_path << " failed.");
			return false;
		}
		fs::close_guard file_guard(_file); // 保证退出时关闭文件
		
		table_schema::pointer schema = this->get_schema();
		if (! schema)
		{
			LOG(ERROR, "tablet::add table schema of '"
			    << this->table_name() << "' is null.");
			return false;
		}

		// 使用较大的缓存区，一次读取较多的数据；并且只在剩余数
		// 据可能不足一条日志数据时，才再次从文件中读取。
		//
		// 这样可明显减少文件读取次数，提高加载日志文件的效率
		basic_simple_buffer<default_char_allocator> file_buffer;
		const std::size_t buffer_size = 8ULL * 1024ULL * 1024ULL; // 8M
		file_buffer.prepare(buffer_size);

		log_entry_v2_for_load entry;
		std::size_t need_bytes = 0;
		bool is_eof = false;
		bool entry_loaded = false;
		do {
			for(;;) { // 循环加载，调整缓存，直到够一条日志记录为止。
				const bool decode_ok = try_decode_log_entry_v2(entry, need_bytes, file_buffer);
				if(! decode_ok) // error
				{
					const boost::asio::const_buffer data = file_buffer.data();
					const char *data_begin = boost::asio::buffer_cast<const char*>(data);
					std::size_t dump_size = boost::asio::buffer_size(data);
					bool has_more = false;
					if(dump_size > 1024)
					{
						dump_size = 1024;
						has_more = true;
					}
					LOG(ERROR, "load_oldlog " << p_path
					    << " failed: decode log entry failed, data hex = "
					    << dump_string_hex(data_begin, dump_size)
					    << (has_more ? " ... " : ""));
					return false;
				}
				else if(need_bytes == 0) // enough
				{
					entry_loaded = true;
					break;
				}
				else // not enough
				{
					if (is_eof)
					{
						if(file_buffer.size() > 0)
						{
							LOG(WARN, "tablet::load_oldlog last log entry is bad(has "
							    << file_buffer.size() << " bytes and need "
							    << need_bytes << "bytes), ignore it");
							file_buffer.clear();
						}
						// 如果此时缓存区中没有数据，则是所有数据都正好处理完了，跳出即可
						entry_loaded = false;
						break;
					}

					// 缓存区没满，则尽量填满
					const std::size_t free_buffer_size = file_buffer.max_size() - file_buffer.size();
					const std::size_t read_bytes = (std::max)(need_bytes, free_buffer_size);

					const boost::asio::mutable_buffer buf = file_buffer.prepare(read_bytes);
					const fs::ssize_t _len = fs::readn(_file, buf);
					if(_len < 0) // error
					{
						LOG(ERROR, "load_oldlog " << p_path
						    << " failed: read file error, fs::errno = "
						    << fs::get_errno());
						return false;
					}
					else if(static_cast<std::size_t>(_len) < read_bytes) // eof
					{		       // ＞0的有符号数转换为无符号数更安全一些。无符号数转换为有符号数可能会溢出成为负值。
						is_eof = true; // also success
					}
					file_buffer.commit(_len);
				}
			}

			// data len is zero, so it's true file is read over
			//if(file_buffer.size() == 0) {
			if(! entry_loaded) { // 文件读完了
				break;
			}

			// deal with this log entry
			//const log_entry_wrapper *entry = wrapper_cast<const log_entry_wrapper*>(file_buffer.data());
			//file_buffer.consume(entry->bytes()); // 必须在此设置

			if(! entry.has_data())
				continue;

			// 检查子表名
			if (entry.tablet_name() != this->tablet_name())
			{
				LOG(WARN, "tablet::load_oldlog log entry (tablet_name:"
				    << dump_string_char(entry.tablet_name())
				    << ", sequence_number:"
				    << entry.m_sequence_number
				    << ")"
				    << " does not match tablet name "
				    << this->tablet_name()
				    << ", skip it.");
				continue;
			}

			// 检查行名
			const std::string &row_name = entry.row_name();
			if(! IS_IN_RANGE(row_name, m_start_row, m_end_row))
			{
				LOG(WARN, "tablet::load_oldlog log entry (row name:"
				    << dump_string_char(row_name)
				    << ")"
				    << " does not in range("
				    << dump_string_char(this->m_start_row)
				    << ","
				    << dump_string_char(this->m_end_row)
				    << "), skip it.");
				continue;
			}
			
			// make row block
			typedef std::pair<group::pointer, row_block::pointer> group_row_block;
			std::vector<group_row_block> row_blocks;
			row_blocks.reserve(entry.m_data.m_groups.size());
			BOOST_FOREACH(const protocol::row_mutation_data_for_tabletserver::group &g, entry.m_data.m_groups)
			{
				
				if(boost::asio::buffer_size(g.second) == 0) // skip empty data
					continue;
				
				if(! schema->has_group(g.first)) // 组可能已经被删除了，所以是正常情况
				{
					LOG(WARN, "tablet::load_oldlog group "
					    << g.first
					    << " does not exists, skip");
					continue;
				}
				
				group::pointer p = get_or_add_group(g.first);
				if(! p)
				{
					LOG(ERROR, "tablet::load_oldlog get or add group "
					    << g.first << " failed");
					return false;
				}
								
				const row_block::pointer row = make_row_block(g.second);
				row_blocks.push_back(group_row_block(p, row));
			}

			if(row_blocks.empty())
			{
				continue;
			}

			// 恢复到各组
			BOOST_FOREACH(group_row_block &row, row_blocks)
			{
				row.first->recover(row.second);
			}

		} while((! is_eof) // 文件没读完
			|| (file_buffer.size() > 0)); // 或缓存数据没读完

		LOG(TRACE, "load_oldlog " << p_path << " recovered");

		return true;
	}

	// 返回false表示其中有一个失败，但保证每个组都进行过一次紧缩；
	// 对运行时来说，紧缩失败可能没有关系（？）
	//
	// 执行时，调用者要根据情况，对子表加读锁
	bool tablet::minor_compact(bool p_forced, bool &need_major_compact)
	{
		//Guard<ReadWriteLock::ReadLock> guard(m_groups_lock.read_lock());
		boost::shared_lock<boost::shared_mutex> guard(m_groups_lock);
		bool ok = true;
		need_major_compact = false;
		BOOST_FOREACH(group_map::value_type &_elem, m_groups)
		{
			const bool ret = _elem.second->minor_compact(p_forced);
			// 更新标志
			need_major_compact = (need_major_compact
					      || _elem.second->need_major_compact());
			ok = ok && ret;
		}
		return ok;	// 只要有一个失败，则返回false
	}

	// 专门用于恢复后的紧缩
	bool tablet::recover_compact()
	{
		//Guard<ReadWriteLock::ReadLock> guard(m_groups_lock.read_lock());
		boost::shared_lock<boost::shared_mutex> guard(m_groups_lock);
		bool ok = true;
		BOOST_FOREACH(group_map::value_type &_elem, m_groups)
		{
			bool ret = _elem.second->recover_compact();
			ok = ok && ret;
		}
		return ok;	// 只要有一个失败，则返回false
	}

	// 在这里保证，对根子表，永远不进行分裂！
	// 调用时要对子表加写锁！
	bool tablet::major_compact_or_split()
	{
		LOG(INFO, "tablet::major_compact_or_split starting on tablet "
		    << tablet_name());
		LOG_ENV(const uint32_t debug_time_start = cur_seconds());

		// 1. set status to be non-available
		// close tablet for a while

		// 由调用者设置为下线状态
		//const bool _origin_online_status = set_online(false);

		// 2. do compact in each group

		std::vector<group::pointer> _compacted;
		std::vector<group::pointer> _uncompacted;
		std::vector<std::string> _split_row_list;
		const bool not_root_meta_tablet = (! this->is_root_meta_tablet());
		size_t _need_split = 0;
		bool _split_ok = true;
		{
			//Guard<ReadWriteLock::ReadLock> guard(m_lock.read_lock());
			boost::shared_lock<boost::shared_mutex> guard(m_lock);
			
			{
				//Guard<ReadWriteLock::ReadLock> _guard(m_groups_lock.read_lock());
				boost::shared_lock<boost::shared_mutex> _guard(m_groups_lock);
				BOOST_FOREACH(group_map::value_type &_elem, m_groups)
				{
					// 主要紧缩完成后，检查是否需要进行分裂
					if (_elem.second->need_major_compact())
					{
						_elem.second->major_compact();
						_compacted.push_back(_elem.second);
					}
					else
					{
						_uncompacted.push_back(_elem.second);
					}
					
					// test split - 当且仅当不是根子表时，才检查是否需要分裂；根子表永远不分裂！
					//
					if(not_root_meta_tablet)
					{
						// 如果本来只有一个SSTable，且较大（很
						// 可能是由于上次分裂没成功完成导致），
						// 这时候就不会进行主要紧缩，但需要进行
						// 分裂，所以放在这里进行检查
						//
						// 如果有多个SSTable文件，会返回false
						if(_elem.second->is_need_split())
						{
							++ _need_split;
							// 只取超过最大大小的SSTable的中间行
							//
							// 这时该组下只有一个SSTable
							const std::string mid_row = _elem.second->get_mid_row();
							if(! mid_row.empty())
							{
								assert(IS_IN_RANGE(mid_row, m_start_row, m_end_row));
								if(mid_row != m_start_row)
								{
									_split_row_list.push_back(mid_row);
								}
								else // 该分裂行无效
								{
									LOG(INFO, "tablet::major_compact_or_split skip invalid mid row "
									    << mid_row);
								}
							}
						}
					}
				}
			}
			
			// 如果是根子表，则 _need_split 一定等于0,所以总会返回 true
			
			set_split(false);    // clear split status
			if (_need_split > 0) // 有一个组需要分裂，就都分裂
			{
				// 将没有分裂的进行强制分裂，确保每个组下只有一个SSTable了
				BOOST_FOREACH(group::pointer _g, _uncompacted)
				{
					_g->major_compact(); // 强制主要紧缩
					if (_g->is_need_split()) {
						// 只取超过最大大小的SSTable的中间行
						//
						// 这时该组下只有一个SSTable
						const std::string mid_row = _g->get_mid_row();
						if(! mid_row.empty())
						{
							assert(IS_IN_RANGE(mid_row, m_start_row, m_end_row));
							if(mid_row != m_start_row)
							{
								_split_row_list.push_back(_g->get_mid_row());
							}
							else // 无效的分裂行
							{
								LOG(INFO, "tablet::major_compact_or_split skip invalid mid row "
								    << mid_row);
							}
							
						}
					}
				}

				// 
				// 然后分裂，计算分裂行并保存，生成两个子表，并
				// 为其创建子表目录和组目录，并在每个组目录中创
				// 建指向当前组目录中SSTable的链接文件。
				//
				
				std::sort(_split_row_list.begin(), _split_row_list.end()); // 排序中间行
				if(_split_row_list.empty()) // 需要计算分裂行
				{
					const std::string mid_row = calc_split_row(m_start_row, m_end_row);
					this->split_row(mid_row);
					LOG(WARN, "tablet::major_compact_or_split get mid row failed, use "
					    << mid_row << " instead");
				}
				else
				{
					// 取中间值作为最终的中间行来分裂
					const std::string &mid_row = _split_row_list[_split_row_list.size() >> 1];
					this->split_row(mid_row);
					LOG(INFO, "tablet::major_compact_or_split to be split by "
					    << mid_row);
				}
			}
		}
		// 释放读锁

		// 到此，需要紧缩的组已经紧缩了
		LOG_ENV(const uint32_t debug_time_middle = cur_seconds());

		// 上面准备好后，在这里真正进行分裂
		if(_need_split > 0)
		{
			// split中会加上写锁，并置为下线状态
			const bool _ok = this->split();
			_split_ok = _ok;
			LOG_IF_ELSE(_ok,
				    INFO, "tablet::major_compact_or_split tablet split successfully",
				    ERROR, "tablet::major_compact_or_split split tablet failed, abort");
		}

		// 再由调用者恢复
		// 3. restore online status
		//set_online(_origin_online_status);

		// 到此，检查完分裂，或许也将其它不需要紧缩的组也紧缩了
		LOG_ENV(const uint32_t debug_time_end = cur_seconds());

		LOG(INFO, "tablet::major_compact_or_split done, normal compacting spent "
		    << debug_time_middle - debug_time_start << "s, compacting-and-splitting spent "
		    << debug_time_end - debug_time_middle << "s");

		return _split_ok;
	}

	bool tablet::split()
	{
		// 让子表下线
		{
			//Guard<ReadWriteLock::WriteLock> guard(m_lock.write_lock());
			boost::unique_lock<boost::shared_mutex> guard(m_lock);
			if(! is_online())
			{
				LOG(ERROR, "tablet::split tablet " << tablet_name()
				    << " is not online");
				return false;
			}

			// 分裂期间仍然可以扫描，但分裂完成后，子表被删除后，扫描就不允许进行了
			// LOG_IF((! not_used()),
			//        WARN,
			//        "tablet::split " << m_scanner_count.value()
			//        << " scanner exist, ignore it");

			set_online(false);
		}

		// 1. 为两个子表生成子表目录
		const protocol::tablet_info split_a_info = get_split_a_info();
		const protocol::tablet_info split_b_info = get_split_b_info();
		const std::string split_a_name = split_a_tablet_name();
		const std::string split_b_name = split_b_tablet_name();
		const std::string path_a = naming_strategy::tablet_path(global::xbase_root_path().string(),
									split_a_info.m_table_name,
									split_a_info.m_tablet_id,
									split_a_info.m_start_row,
									split_a_info.m_end_row);
		const std::string path_b = naming_strategy::tablet_path(global::xbase_root_path().string(),
									split_b_info.m_table_name,
									split_b_info.m_tablet_id,
									split_b_info.m_start_row,
									split_b_info.m_end_row);
		bool _ok = fs::mkdir(path_a);
		if(! _ok) {
			LOG(ERROR, "create split_a tablet directory "
			    << path_a
			    << " failed.");
			return false;
		}
		_ok = fs::mkdir(path_b);
		if(! _ok) {
			LOG(ERROR, "create split_b tablet directory "
			    << path_b
			    << " failed.");
			
			fs::remove(path_a); // remove path a
			return false;
		}

		// 2. 遍历每个组进行分裂
		const uint64_t _sstable_sequence_number = cur_micro_seconds();
		//Guard<ReadWriteLock::ReadLock> _guard(m_groups_lock.read_lock());
		boost::shared_lock<boost::shared_mutex> _guard(m_groups_lock);
		BOOST_FOREACH(group_map::value_type &_elem, m_groups)
		{
			bool _ok = _elem.second->split(this->split_row(),
						       path_a,
						       m_split_a_id,
						       split_a_name,
						       path_b,
						       m_split_b_id,
						       split_b_name,
						       _sstable_sequence_number);
			if (! _ok) {
				LOG(ERROR, "tablet::split error occur in group "
				    << _elem.second->group_name() << " splitting, abort.");
				fs::remove(path_a); // remove path a
				fs::remove(path_b); // remove path b
				return false;
			}
		}

		// 设置分裂状态，外部检测到后会更新元数据表
		set_split(true);
		return true;
	}

	uint64_t tablet::max_sequence_number() const
	{
		//Guard<ReadWriteLock::ReadLock> guard(m_groups_lock.read_lock());
		boost::shared_lock<boost::shared_mutex> guard(m_groups_lock);
		uint64_t max_sequence_number = 0;
		BOOST_FOREACH(const group_map::value_type &_elem, m_groups)
		{
			max_sequence_number = (std::max)(max_sequence_number, _elem.second->max_sequence_number());
		}
		return max_sequence_number;
	}

	void tablet::clean_split_path() const {
		if(m_split_row.empty()) // no split row
		{
			return;
		}

		const protocol::tablet_info split_a_info = get_split_a_info();
		const protocol::tablet_info split_b_info = get_split_b_info();
		const std::string path_a = naming_strategy::tablet_path(global::xbase_root_path().string(),
									split_a_info.m_table_name,
									split_a_info.m_tablet_id,
									split_a_info.m_start_row,
									split_a_info.m_end_row);
		const std::string path_b = naming_strategy::tablet_path(global::xbase_root_path().string(),
									split_b_info.m_table_name,
									split_b_info.m_tablet_id,
									split_b_info.m_start_row,
									split_b_info.m_end_row);
		const bool ok_a = fs::remove(path_a);
		LOG_IF_ELSE((! ok_a),
			    WARN,
			    "tablet::clean_split_path remove split_a path "
			    << path_a << " failed, fs::errno = " << fs::get_errno(),
			    INFO,
			    "tablet::clean_split_path removed split_a path "
			    << path_a);
		(void)ok_a;

		const bool ok_b = fs::remove(path_b);
		LOG_IF_ELSE((! ok_b),
			    WARN,
			    "tablet::clean_split_path remove split_b path "
			    << path_b << " failed, fs::errno = " << fs::get_errno(),
			    INFO,
			    "tablet::clean_split_path removed split_b path "
			    << path_b);
		(void)ok_b;
	}

	
	void tablet::notify_minor_compact() {
		boost::mutex::scoped_lock lock(m_compact_status_lock);
		const bool ret = (m_compact_status == cs_null);
		if(ret)	// 仅当状态位为空（即还没有通知，也没有在做其它紧缩操作），才通知
		{
			global::notify_minor_compact(this->shared_from_this());
			m_compact_status = cs_minor_compact_reported;
		}
		LOG_IF((! ret),
		       DEBUG,
		       "tablet::notify_minor_compact status error " << m_compact_status
		       << ", tablet " << tablet_name());
	}
	
	void tablet::notify_major_compact() {
		boost::mutex::scoped_lock lock(m_compact_status_lock);
		const bool ret = ((m_compact_status == cs_doing_minor_compact)
				  || (m_compact_status == cs_null));
		LOG_IF((! ret),
		       INFO,
		       "tablet::notify_major_compact status error " << m_compact_status
		       << ", tablet " << tablet_name());
		if(ret)	// 仅当状态位为空（一开始就发现需要主要紧缩）或正在次要紧缩（次要紧缩后检查需要主要紧缩），才通知
		{
			global::notify_major_compact(this->shared_from_this());
			m_compact_status = cs_major_compact_reported;
		}
	}

	void tablet::check_compact() {
		boost::mutex::scoped_lock lock(m_compact_status_lock);
		if(m_compact_status == cs_null)
		{
			if(is_need_minor_compact())
			{
				global::notify_minor_compact(this->shared_from_this());
				m_compact_status = cs_minor_compact_reported;
			}
			else if(is_need_major_compact())
			{
				global::notify_major_compact(this->shared_from_this());
				m_compact_status = cs_major_compact_reported;
			}
			// else, do nothing
		}
		else
		{
			LOG(INFO, "tablet::check_compact skip check on tablet " << tablet_name()
			    << "against compact_status " << m_compact_status);
		}
	}

} // namespace xbase
